package com.lihh.rocket5.transaction;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

public class TransactionConsumer {
  public static void main(String[] args) throws MQClientException {
    // 拿到consumer 实例
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_consumer");
    // 设置 name server
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅主题
    consumer.subscribe("TransactionTopic", "*");
    // 设置消费主题
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    // 设置消费模式
    consumer.setMessageModel(MessageModel.CLUSTERING);
    
    // 注册消费监听器
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for (MessageExt msg : list) {
          SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
          System.out.println("begin");
          System.out.println("B服务器本地事务  ,TransactionId：" + msg.getTransactionId());
          System.out.println("update B ...(B账户加100块) :" + df.format(new Date()));
          System.out.println("commit " + df.format(new Date()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    
    consumer.start();
    while (true) {
    }
  }
}
